package com.v.im.imservice.kafka.producer;

import com.v.im.commonmodule.utils.chat.Message;
import com.v.im.commonmodule.utils.kafka.KafkaTopic;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

@Slf4j
@Component
public class ImKafkaProducer  {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 发送单聊数据
     */

    public void sendFriendMessage(Message message){
        log.info("向kafka推送数据:[{}]", message);
        try {
            kafkaTemplate.send(KafkaTopic.FRIEND_TOPIC, message);
        } catch (Exception e) {
            log.error("发送数据出错！！！{}{}", KafkaTopic.FRIEND_TOPIC, message);
            log.error("发送数据出错=====>", e);
        }

        //消息发送的监听器，用于回调返回信息
        kafkaTemplate.setProducerListener(new ProducerListener<String, Message>() {

            @Override
            public void onSuccess(String topic, Integer partition, String key, Message value, RecordMetadata recordMetadata) {
                log.info("kafka 发送数据成功");

            }


            @Override
            public void onError(String topic, Integer partition, String key, Message value, Exception exception) {
                log.error("kafka 发送数据异常");
            }

            @Override
            public boolean isInterestedInSuccess() {
                //log.info("数据发送成功！！！");
                return false;
            }
        });
    }

    /**
     * 发送群聊数据
     */

    public void sendCrowdMessage(Message message){
        ListenableFuture future = kafkaTemplate.send(KafkaTopic.CROWD_TOPIC, message);
        future.addCallback(o -> System.out.println("send-消息发送成功：" + message),
                throwable -> System.out.println("消息发送失败：" + message));
    }
}
